package cn._51doit.live.udfs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.utils.EventType;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


public class AudienceCountFunction extends KeyedProcessFunction<String, DataBean, Tuple3<String, Integer, Integer>> {

    private transient MapState<String, Integer> mapState;

    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor<String, Integer> stateDescriptor = new MapStateDescriptor<>("count-map-state", String.class, Integer.class);
        mapState = getRuntimeContext().getMapState(stateDescriptor);
    }

    @Override
    public void processElement(DataBean value, Context ctx, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
        String eventId = value.getEventId();

        Integer totalCount = mapState.get("TOTAL_COUNT");
        Integer onlineCount = mapState.get("ONLINE_COUNT");

        //进入
        if(EventType.LIVE_ENTER.equals(eventId)) {
            //1.统计累计观众
            if (totalCount == null) {
                totalCount = 0;
            }
            totalCount += 1;
            mapState.put("TOTAL_COUNT", totalCount);
            //2.统计在线用户
            onlineCount +=1;

        } else {
            if (onlineCount == null) {
                onlineCount = 0;
            }
            //离开
            onlineCount -=1;
        }
        mapState.put("ONLINE_COUNT", onlineCount);
        //输出数据
        out.collect(Tuple3.of(ctx.getCurrentKey(), totalCount, onlineCount));

    }


}
